package my.socket;

import my.ByteBufferAllocator;
import my.EpollEventLoop;
import my.common.File;
import my.pipeline.DefaultChannelPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.ByteBuffer;

/**
 * @author fengbo
 */
public class Socket extends AbstractSocket {

    private static final Logger log = LoggerFactory.getLogger(EpollEventLoop.class);

    private final DefaultChannelPipeline pipeline;
    private EpollEventLoop eventLoop;

    public Socket(int fileDescriptor) {
        super(fileDescriptor);
        pipeline = newChannelPipeline();
    }

    public final void register(EpollEventLoop eventLoop) {
        this.eventLoop = eventLoop;
        eventLoop.execute(() -> {
            try {
                eventLoop.add(this);
            } catch (IOException e) {
                log.error("Add task failed, ex=[{}]", e.getMessage(), e);
                this.close();
            }
        });
    }

    public EpollEventLoop getEventLoop() {
        return eventLoop;
    }


    private DefaultChannelPipeline newChannelPipeline() {
        return new DefaultChannelPipeline(this);
    }

    public DefaultChannelPipeline pipeline() {
        return pipeline;
    }

    public void inReady() {
        final DefaultChannelPipeline pipeline = pipeline();

        int limit = 1024;
        ByteBufferAllocator allocator = eventLoop.allocator();
        ByteBuffer buf = allocator.allocate(limit);
        for (;;) {
            if (!buf.hasRemaining()) {
                // 乘以2
                limit <<= 1;
                ByteBuffer oldBuf = buf;
                buf = allocator.allocate(limit);
                allocator.release(buf, limit);
                int position = oldBuf.position();
                oldBuf.flip();
                buf.put(oldBuf);
                buf.position(position);
            }
            int read = File.read(fileDescriptor, buf, buf.position(), buf.limit());
            if (read < 0) {
                break;
            }
            buf.position(buf.position() + read);
        }
        buf.flip();
        pipeline.fireChannelRead(buf);
    }

    public void outReady() {
        String response = Thread.currentThread().getName() + " response\n";
        ByteBuffer buf = ByteBuffer.allocateDirect(response.length());
        buf.put(response.getBytes());
        buf.flip();
        File.write(fileDescriptor, buf, buf.position(), buf.limit());
        buf.clear();
    }
}
